package streamAPI.checkpoint;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

/**
 * 这个是at latestOnce 的原因是 因为 如果我在checkpoint 还没做的时候如果 重启了 那么会重复读
 * @author yue.cao
 * @since 10-27-2020
 */
public class MyAtLatestOnceParallelFileSource extends RichParallelSourceFunction<Tuple2<Integer, String>> implements CheckpointedFunction {

	private String path;

	private boolean flag = true;

	private long offset = 0;

	private int subTaskIndex;

	private ListState<Long> offsetState;

	public MyAtLatestOnceParallelFileSource(String path) {
		this.path = path;
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
	}

	@Override
	public void run(SourceContext<Tuple2<Integer, String>> ctx) throws Exception {
		RandomAccessFile randomAccessFile = new RandomAccessFile(path + "/" + subTaskIndex, "r");
		randomAccessFile.seek(offset);

		final Object checkpointLock = ctx.getCheckpointLock();
		while (flag) {
			String line = randomAccessFile.readLine();
			if (line != null) {
				line = new String(line.getBytes("ISO-8859-1"), "UTF-8");
				// 希望在做checkpoint的时候停下
				synchronized (checkpointLock) {
					offset = randomAccessFile.getFilePointer();
					ctx.collect(Tuple2.of(subTaskIndex, line));
				}
			} else {
				TimeUnit.SECONDS.sleep(1);
			}
		}

	}

	@Override
	public void cancel() {
		flag = false;
	}

	/**
	 * 定期将状态数据保存到stateBackend
	 */
	@Override
	public void snapshotState(FunctionSnapshotContext context) throws Exception {
		offsetState.clear();
		offsetState.add(offset);
	}

	/**
	 * 初始化 OperatorState 生命周期方法 构造方法执行后 会执行一次
	 */
	@Override
	public void initializeState(FunctionInitializationContext context) throws Exception {
		ListStateDescriptor<Long> descriptor = new ListStateDescriptor<>(
				"file-offset-state",
//				TypeInformation.of(new TypeHint<Long>() {})
//				Types.LONG
				Long.class
		);
		offsetState = context.getOperatorStateStore().getListState(descriptor);
		subTaskIndex = getRuntimeContext().getIndexOfThisSubtask();
		if (context.isRestored()) {
			for (Long first : offsetState.get()) {
				offset = first;
			}
		}
	}
}
